
#include "DataTransmissionTcp.h"
#include "WingsTcpUdpProto.h"
#include "Proc.h"
#include "Conf.h"
#include "Misc.h"
#include "Crc.h"
#include "myLog.h"



using namespace std;

DataTransmissionTcp::~DataTransmissionTcp() 
{
    this->deinit();
}

Result DataTransmissionTcp::init(const initializer_list<void *> attr) 
{
    m_localPort = Conf::currentConf()->tcp.port;
    m_headerSize = Conf::currentConf()->tcp.headerSize;
    m_type = Conf::currentConf()->typeValue;

    m_sock = ::socket(AF_INET, SOCK_STREAM, 0);
    if (m_sock < 0) 
    {
        int err = errno;
        auto info = getErrorInfo(err);
        auto res = Result::create(ErrorKind::InitFailed);
        res.appendInfo("tcp data transmission, create tcp socket failed");
        res.appendInfo(info);

        return res;
    }

    auto exitClosure = [&](const Result &res){
        if (::close(this->m_sock) < 0) 
        {
            int err = errno;
            qlog_e("tcp data transmission, close tcp socket failed, detail: %s", getErrorInfo(err).c_str());
        }
        this->m_sock = -1;

        return res;
    };

    // set recv/accept timeout.
    timeval timeout = { 0 };
    timeout.tv_sec = 3;
    timeout.tv_usec = 0;
    auto ret = setsockopt(m_sock, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeval));
    if (ret < 0) 
    {
        int err = errno;
        auto info = getErrorInfo(err);
        auto res = Result::create(ErrorKind::ExecuteFailed);
        res.appendInfo("tcp data transmission, set tcp socket recv timeout failed");
        res.appendInfo(info);

        return exitClosure(res);
    }

    int bufSize = 256 * 1024;
    ret = setsockopt(m_sock, SOL_SOCKET, SO_RCVBUF, &bufSize, sizeof(int));
    if (ret < 0) 
    {
        int err = errno;
        auto info = getErrorInfo(err);
        auto res = Result::create(ErrorKind::ExecuteFailed);
        res.appendInfo("tcp data transmission, set tcp socket recv buffer size failed");
        res.appendInfo(info);

        return exitClosure(res);
    }

    // bind
    sockaddr_in localAddr = { 0 };
    localAddr.sin_family = AF_INET;
    localAddr.sin_addr.s_addr = INADDR_ANY;
    localAddr.sin_port = htons(m_localPort);
    ret = ::bind(m_sock, reinterpret_cast<sockaddr *>(&localAddr), sizeof(sockaddr_in));
    if (ret < 0) 
    {
        int err = errno;
        auto info = getErrorInfo(err);
        auto res = Result::create(ErrorKind::ExecuteFailed);
        res.appendInfo("tcp data transmission, bind tcp socket failed");
        res.appendInfo(info);

        return exitClosure(res);
    }

    ret = ::listen(m_sock, sizeof(uint8_t));
    if (ret < 0) 
    {
        int err = errno;
        auto info = getErrorInfo(err);
        auto res = Result::create(ErrorKind::ExecuteFailed);
        res.appendInfo("tcp data transmission, listen tcp socket failed");
        res.appendInfo(info);

        return exitClosure(res);
    }

    m_workerExitFlag = false;
    m_worker = thread(&DataTransmissionTcp::run, this);
    setThreadName("tcp background");

    return Result();
}

Result DataTransmissionTcp::transmit(const void *src, const size_t size) 
{
    if (nullptr == src || 0 == size) 
    {
        auto res = Result::create(ErrorKind::InvalidArgument);
        res.appendInfo("tcp data transmission, passed invalid arguments to transmit function");

        return res;
    }

    if (m_peerSock < 0) 
    {
        auto res = Result::create(ErrorKind::InvalidOperation);
        res.appendInfo("tcp data transmission, invalid peer socket.");

        return res;
    }

    auto ret = ::send(m_peerSock, src, size, 0);
    if (ret <= 0) 
    {
        int err = errno;
        auto info = getErrorInfo(err);
        auto res = Result::create(ErrorKind::ExecuteFailed);
        res.appendInfo("tcp data transmission, transmit failed, detail: ");
        res.appendInfo(info);

        return res;
    }
    
    if (size != static_cast<size_t>(ret)) 
    {
        return Result::create(ErrorKind::ImcompleteIo, "tcp data transmission, sent bytes is not equal to data size");
    }

    return Result();
}

void DataTransmissionTcp::release(const void *item) 
{
    if (nullptr == item) 
    {
        return;
    }
    auto ptr = const_cast<WingsInternalWrapper *>(reinterpret_cast<const WingsInternalWrapper *>(item));
    delete ptr;
}

void DataTransmissionTcp::deinit()  
{
    if (false == m_workerExitFlag) 
    {
        m_workerExitFlag = true;

        if (true == m_worker.joinable()) 
        {
            m_worker.join();
        }
        else 
        {
            m_worker.detach();
        }
    }

    if (this->m_peerSock >= 0 && ::close(this->m_peerSock) < 0) 
    {
        int err = errno;
        qlog_e("tcp data transmission, close peer tcp socket failed(deinit), detail: %s", getErrorInfo(err).c_str());
    }
    this->m_peerSock = -1;

    if (::close(this->m_sock) < 0) 
    {
        int err = errno;
        qlog_e("tcp data transmission, close tcp socket failed(deinit), detail: %s", getErrorInfo(err).c_str());
    }
    this->m_sock = -1;

}

void DataTransmissionTcp::run() 
{
    qlog_i("wings tcp thread start.");
    vector<uint8_t> buffer;
    int ret = -1;
    const uint16_t header = 0x2014, tail = 0x0022;
    uint32_t bodySize = 0, nameSize = 0, remain = 0;
    uint32_t crc32 = 0;
    uint32_t tmp32 = 0;
    uint16_t tmp16 = 0;

    // 状态字符串 + ',' + 状态码, 状态码是0表示处理完毕，-1表示数据域语义有错误(无效的数据)，-2表示crc校验错误，其他值是未定义的。
    // 举例: 
    // "header,-1" 表示接收数据包的包头失败,数据无效。接收包头正常时，不会回复此条信息。
    // "body,-1" 表示接收数据包的包体失败,数据无效。接收包体正常时，不会回复此条信息。
    // "process,-1" 表示接收完毕，但处理数据包的时候有语义错误。
    // "process,0" 表示接收完毕，且处理成功。
    auto response = [&](const char *stat, const int value){
        string msg = stat;
        msg += ",";
        msg += to_string(value);
        msg += "\n";

        auto res = this->transmit(msg.data(), msg.size());
        if (false == res.isOk()) 
        {
            qlog_e("response failed, %s", res.m_message.c_str());
            return false;
        }

        return true;
    };

    auto next = [&]{
        if (this->m_peerSock >= 0) 
        {
            if (::close(this->m_peerSock) < 0) 
            {
                int err = errno;
                qlog_e("tcp data transmission, close peer tcp socket failed(next), detail: %s", getErrorInfo(err).c_str());
            }
            this->m_peerSock = -1;
        }
    };

    while (true) 
    {
        if (true == m_workerExitFlag) 
        {
            break;
        }

        m_peerSock = ::accept4(m_sock, reinterpret_cast<sockaddr *>(&m_peerAddr), &m_peerAddrSize, 0);
        if (m_peerSock < 0) 
        {
            int err = errno;
            if (EWOULDBLOCK != err) 
            {
                qlog_e("tcp data transmission, accept failed, detail: %s", getErrorInfo(err).c_str());
            }
            
            continue;
        }

        qlog_v("accept new connection, peer(%s:%d).", inet_ntoa(m_peerAddr.sin_addr), ntohs(m_peerAddr.sin_port));

        // recv header.
        buffer.resize(m_headerSize, 0);
        ret = ::recv(m_peerSock, buffer.data(), buffer.size(), MSG_WAITALL);
        if (ret <= 0) 
        {
            int err = errno;
            if (EWOULDBLOCK != err) 
            {
                qlog_e("tcp data transmission, recv header failed, detail: %s", getErrorInfo(err).c_str());
            }

            next();
            continue;
        }

        qlog_v("recv %d bytes header from '%s:%d'", ret, inet_ntoa(m_peerAddr.sin_addr), ntohs(m_peerAddr.sin_port));
        if ((int)m_headerSize != ret) 
        {
            qlog_e("tcp data transmission, header size is %d bytes but recv %d bytes", m_headerSize, ret);
            response("header", -1);

            next();
            continue;
        }
        //提取帧头
        tmp16 = static_cast<uint16_t>(buffer[1]) << 8 | static_cast<uint16_t>(buffer[0]);
        if (WINGS_HEADER != tmp16) 
        {
            qlog_e("tcp data transmission, invalid header.");
            response("header", -1);

            next();
            continue;
        }
        //提取设备类型
        tmp32 = static_cast<uint32_t>(buffer[7]) << 24 | static_cast<uint32_t>(buffer[6]) << 16 | static_cast<uint32_t>(buffer[5]) << 8 | static_cast<uint32_t>(buffer[4]);
        if (m_type != tmp16) 
        {
            qlog_e("tcp data transmission, invalid type '%d', it's not for me.", *(reinterpret_cast<decltype(header) *>(buffer.data()) + 1));
            response("header", -1);

            next();
            continue;
        }
        //提取数据包大小
        bodySize = static_cast<uint32_t>(buffer[11]) << 24 | static_cast<uint32_t>(buffer[10]) << 16 | static_cast<uint32_t>(buffer[9]) << 8 | static_cast<uint32_t>(buffer[8]);
        tmp32 = static_cast<uint32_t>(buffer[15]) << 24 | static_cast<uint32_t>(buffer[14]) << 16 | static_cast<uint32_t>(buffer[13]) << 8 | static_cast<uint32_t>(buffer[12]);
        if (WINGS_TAIL != tmp32) 
        {
            qlog_e("tcp data transmission, invalid tail");
            response("header", -1);

            next();
            continue;
        }
        if(bodySize > 0)
        {
            //数据包=数据包名字+数据实体
            // recv remaining.
            buffer.reserve(m_headerSize + bodySize);
            remain = bodySize;
            while(true) 
            {
                if (remain <= 0) 
                {
                    response("body size", bodySize);
                    break;
                }

                ret = ::recv(m_peerSock, buffer.data() + m_headerSize, remain, MSG_WAITALL);
                if (ret <= 0) 
                {
                    int err = errno;
                    qlog_e("tcp data transmission, recv body failed, detail: %s", getErrorInfo(err).c_str());
                    break;
                }

                remain -= ret;
            }
        }
        
        if (remain > 0) 
        {
            qlog_e("tcp data transmission, data received is imcomplete.");
            response("body error", -1);

            next();
            continue;
        }

        auto wrapper = new(WingsInternalWrapper);
        auto reply = this->dataResolver(move(buffer),*wrapper);
        if(reply != nullptr) {
            int size = reinterpret_cast<WingsUpdateHeaderTcpUdp *>(reply)->dataSize + m_headerSize;
            this->transmit(reply,size);
        }
        if(!wrapper->name.empty()) {
            this->asyncRecv(wrapper, sizeof(WingsInternalWrapper));
            response("process", 0);
        }

        response("process", 0);
        next();
    }

    qlog_i("wings tcp thread exit.");
}
